package com.hulk.rpc.transport.client.codec;

import com.hulk.rpc.protocol.Response;
import com.hulk.rpc.serialize.Serializer;
import com.hulk.rpc.transport.client.future.FutureContainer;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import lombok.extern.slf4j.Slf4j;

import static com.hulk.rpc.config.common.HulkConstants.HEADER_FIELD_LENGTH;

/**
 * @author Administrator
 */
@Slf4j
public class ResponseDecoder extends LengthFieldBasedFrameDecoder {

  private final Serializer serializer;
  private final FutureContainer futureContainer;

  public ResponseDecoder(int maxFrameLength, Serializer serializer, FutureContainer futureContainer) {
    super(maxFrameLength, 0, HEADER_FIELD_LENGTH, 0, HEADER_FIELD_LENGTH);
    this.serializer = serializer;
    this.futureContainer = futureContainer;
  }

  @Override
  protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
    ByteBuf buffer = (ByteBuf) super.decode(ctx, in);

    if (buffer != null) {
      try {
        Response response = serializer.readResponse(buffer);
        futureContainer.notifyResponse(response);
      } finally {
        buffer.release();
      }
    }

    return null;
  }

  @Override
  public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {

    if (log.isErrorEnabled()) {
      log.error("Exception caught on " + ctx.channel(), cause);
    }

    ctx.channel().close();
    this.futureContainer.close();
  }

}
